package com.heima.kafka.simple;

import org.apache.kafka.clients.producer.*;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * 生成者快速入门
 * 发送消息
 */
public class ProducerStart {

    public static void main(String[] args) {
        //2.设置kafka配置信息
        Properties properties=new Properties();
        //2.1 设置kafka地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.105:9092");
        //2.2设置对key,value进行字符串序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //重试的配置
        properties.put(ProducerConfig.RETRIES_CONFIG,10);
        /*
        1.创建生产者对象
         KafkaProducer(Properties properties)
         properties表示属性的意思，能够配置kafka信息
         */

        KafkaProducer kafkaProducer=new KafkaProducer(properties);

        /*
        3.发送消息
        ProducerRecord(String topic, K key, V value)
        参数1表示消息队列名称
        参数2 表示消息的key
        参数3表示的是消息的内容
         */
        for (int i = 0; i <5 ; i++) {
            if(i%2==0){
                ProducerRecord record=new ProducerRecord("kafka-stream-producer","key"+i,"hello kafka");
                //同步发送
                kafkaProducer.send(record);
            }else{
                ProducerRecord record=new ProducerRecord("kafka-stream-producer","key"+i,"hello itcast");
                //同步发送
                kafkaProducer.send(record);
            }
        }

       /* try {
            RecordMetadata recordMetadata= (RecordMetadata) future.get();
            System.out.println(recordMetadata.offset());
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }*/
        //异步发送
      /*  kafkaProducer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if(exception==null){
                    //执行我们的业务
                    System.out.println(metadata.offset());
                }else{
                    //执行重试机制，重试发送消息
                    System.out.println(exception);
                }
            }
        });*/

        //4.关闭流
        kafkaProducer.close();
    }
}
